package com.shujia.flink.fraud

import org.apache.flink.streaming.api.scala._

object FraudDetectionJob {
  def main(args: Array[String]): Unit = {

    /**
      * 在当今数字时代，信用卡欺诈行为越来越被重视。
      * 罪犯可以通过诈骗或者入侵安全级别较低系统来盗窃信用卡卡号。
      * 用盗得的信用卡进行很小额度的例如一美元或者更小额度的消费进行测试。
      * 如果测试消费成功，那么他们就会用这个信用卡进行大笔消费，
      * 来购买一些他们希望得到的，或者可以倒卖的财物。
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val lines = env.socketTextStream("node1", 5555)

    val event = lines.map(line => {
      val split = line.split(",")
      val id = split(0)
      val amount = split(1).toDouble
      Event(id, amount)
    })

    val keyedStream = event.keyBy(_.id)

    val alertDS = keyedStream.process(new FraudDetector)

    alertDS.print()

    env.execute()


  }

}

case class Event(id: String, amount: Double)
